package com.bff.gaia.mix.api.xjoin2.client.mysql.all;

import com.alibaba.fastjson.JSONObject;
import com.bff.gaia.api.java.tuple.Tuple;
import com.bff.gaia.api.java.tuple.Tuple5;
import com.bff.gaia.api.java.tuple.Tuple6;
import com.bff.gaia.configuration.Configuration;
import com.bff.gaia.mix.api.functions.source.RichSourceFunction;
import com.bff.gaia.mix.api.xjoin2.client.rdb.table.RdbSideTableInfo;
import com.bff.gaia.mix.api.xjoin2.core.side.SideTableInfo;
import com.bff.gaia.mix.api.xjoin2.core.table.FieldInfo;
import com.bff.gaia.mix.api.xjoin2.core.table.JoinInfo;
import com.bff.gaia.mix.api.xjoin2.core.table.StreamTableInfo;
import com.bff.gaia.mix.api.xjoin2.core.table.XjoinInfo;
import com.bff.gaia.mix.api.xjoin2.core.utils.StringUtil;
import com.bff.gaia.shaded.guava18.com.google.common.collect.Maps;
import com.bff.gaia.util.MathUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicReference;

import static com.bff.gaia.mix.api.xjoin2.core.utils.KeyUtil.buildKey;

public class MysqlAllXjoinSourceFunction extends RichSourceFunction<Tuple5<Integer,String,String,String,Integer>> {

	private static final long serialVersionUID = 7346707398402801134L;

	protected XjoinInfo xjoinInfo;

	private static final int CONN_RETRY_NUM = 3;

	private int version = 0;

	private int currentVersion = 0;

	private static final Logger LOG = LoggerFactory.getLogger(MysqlAllXjoinSourceFunction.class);

	private AtomicReference<ConcurrentHashMap<String, ArrayList<String>>> cacheRef2 = new AtomicReference<>();

	private boolean keep = true;

	private long cacheTime;

	private final String MYSQL_DRIVER = "com.mysql.jdbc.Driver";

	private final String MYSQL_DRIVER_6 = "com.mysql.cj.jdbc.Driver";

	public MysqlAllXjoinSourceFunction(StreamTableInfo streamTableInfo, SideTableInfo sideTableInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList) {
		this.xjoinInfo = new MysqlAllXjoinInfo(streamTableInfo, sideTableInfo, joinInfo, outFieldInfoList);
	}

	@Override
	public void open(Configuration parameters) throws Exception {
		super.open(parameters);
		ConcurrentHashMap<String, ArrayList<String>> newCache2 = new ConcurrentHashMap<>();
		this.cacheRef2.set(newCache2);
		this.version = 0;
		this.cacheTime = this.xjoinInfo.getJoinInfo().getCacheTimeout();
	}

	@Override
	public void run(SourceContext<Tuple5<Integer,String,String,String,Integer>> ctx) throws Exception {
		while (keep) {
			loadData(ctx);
			this.version++;
			Thread.sleep(cacheTime);
		}
	}

	@Override
	public void cancel() {
		this.keep = false;
		cacheRef2.get().clear();
	}

	@Override
	public void close() throws Exception {
		super.close();
		cacheRef2.get().clear();
	}

	public Connection getConn(String dbURL, String userName, String password) {
		try {
			Class.forName(MYSQL_DRIVER_6);
			//add param useCursorFetch=true
			Map<String, String> addParams = Maps.newHashMap();
			addParams.put("useCursorFetch", "true");
			String targetDbUrl = StringUtil.addJdbcParam(dbURL, addParams, true);
			return DriverManager.getConnection(targetDbUrl, userName, password);
		} catch (Exception e) {
			LOG.error("", e);
			throw new RuntimeException("", e);
		}
	}

	public int getFetchSize() {
		return Integer.MIN_VALUE;
//		return 10000;
	}

	public void loadData(SourceContext<Tuple5<Integer,String,String,String,Integer>> ctx) throws SQLException {
		System.out.println("----------------begin loadData------------------------");
		ConcurrentHashMap<String, ArrayList<String>> oldCache = this.cacheRef2.get();
		ConcurrentHashMap<String, ArrayList<String>> newCache = new ConcurrentHashMap<>();
		RdbSideTableInfo tableInfo = (RdbSideTableInfo) xjoinInfo.getSideTableInfo();

		Connection connection = null;

		try {
			for (int i = 0; i < CONN_RETRY_NUM; i++) {
				try {
					connection = getConn(tableInfo.getUrl(), tableInfo.getUserName(), tableInfo.getPassword());
					break;
				} catch (Exception e) {
					if (i == CONN_RETRY_NUM - 1) {
						throw new RuntimeException("", e);
					}
					try {
						String connInfo = "url:" + tableInfo.getUrl() + ";userName:" + tableInfo.getUserName() + ",pwd:" + tableInfo.getPassword();
						LOG.warn("get conn fail, wait for 5 sec and try again, connInfo:" + connInfo);
						Thread.sleep(5 * 1000);
					} catch (InterruptedException e1) {
						e1.printStackTrace();
					}
				}
			}

			//load data from table
			String sql = xjoinInfo.getSqlQuery();
			PreparedStatement statement = connection.prepareStatement(
				sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
			statement.setFetchSize(getFetchSize());
			ResultSet resultSet = statement.executeQuery();
			String[] sideFieldNames = xjoinInfo.getSideSelectFields().split(",");

			while (resultSet.next()) {
				JSONObject oneRow = new JSONObject();

				for (String fieldName : sideFieldNames) {
					Object object = resultSet.getObject(fieldName.trim());
					oneRow.put(fieldName.trim(), String.valueOf(object));
				}
				String cacheKey = buildKey(oneRow, xjoinInfo.getSideTableKeyFieldVal());
				String fullKey = oneRow.toJSONString();
				System.out.println("cacheKey: " + cacheKey);
				System.out.println("fullKey: " + fullKey);

				if (xjoinInfo.getJoinInfo().isIncLog()) {
					//add keyInfo to new cache
					ArrayList<String> newFullkeys = newCache.getOrDefault(cacheKey, new ArrayList<String>());
					newFullkeys.add(fullKey);
					newCache.put(cacheKey, newFullkeys);

					//check if data is new
					if (oldCache.containsKey(cacheKey)) {
						//if cacheKey exist
						//find fullkey
						ArrayList<String> fullkeys = oldCache.get(cacheKey);
						if (!fullkeys.contains(fullKey)) {
							//if fullkey do not exist,data is new, need send
//							JSONObject changeLog = new JSONObject();
//							changeLog.put("version", this.version);
//							changeLog.put("action", "add");
//							changeLog.put("cacheKey", cacheKey);
//							changeLog.put("fullKey", fullKey);
//							changeLog.put("value", oneRow);
//							changeLog.put("gateway", MathUtils.murmurHash(cacheKey.hashCode()) % xjoinInfo.getJoinInfo().getParallelism());
							Tuple5<Integer,String,String,String,Integer> changeLog=new Tuple5<Integer,String,String,String,Integer>(
								this.version,
								"add",
								cacheKey,
								fullKey,
								MathUtils.murmurHash(cacheKey.hashCode()) % xjoinInfo.getJoinInfo().getParallelism()
							);
							ctx.collect(changeLog);
						} else {
							//if fullkey exist,data is old, do not change
							//remove fullkey fromoldCache
							fullkeys.remove(fullKey);
							oldCache.put(cacheKey,fullkeys);
						}
						if (fullkeys.isEmpty()) {
							oldCache.remove(cacheKey);
						}
					} else {
						//if cacheKey do not exist
						//data is new,need send
//						JSONObject changeLog = new JSONObject();
//						changeLog.put("version", this.version);
//						changeLog.put("action", "add");
//						changeLog.put("cacheKey", cacheKey);
//						changeLog.put("fullKey", fullKey);
//						changeLog.put("value", oneRow);
//						changeLog.put("gateway", MathUtils.murmurHash(cacheKey.hashCode()) % xjoinInfo.getJoinInfo().getParallelism());
						Tuple5<Integer,String,String,String,Integer> changeLog=new Tuple5<Integer,String,String,String,Integer>(
							this.version,
							"add",
							cacheKey,
							fullKey,
							MathUtils.murmurHash(cacheKey.hashCode()) % xjoinInfo.getJoinInfo().getParallelism()
						);
						ctx.collect(changeLog);
					}
				} else {
//					JSONObject changeLog = new JSONObject();
//					changeLog.put("version", this.version);
//					changeLog.put("action", "add");
//					changeLog.put("cacheKey", cacheKey);
//					changeLog.put("fullKey", fullKey);
//					changeLog.put("value", oneRow);
//					changeLog.put("gateway", MathUtils.murmurHash(cacheKey.hashCode()) % xjoinInfo.getJoinInfo().getParallelism());
					Tuple5<Integer,String,String,String,Integer> changeLog=new Tuple5<Integer,String,String,String,Integer>(
						this.version,
						"add",
						cacheKey,
						fullKey,
						MathUtils.murmurHash(cacheKey.hashCode()) % xjoinInfo.getJoinInfo().getParallelism()
					);
					ctx.collect(changeLog);
				}

			}
		} catch (Exception e) {
			e.printStackTrace();
			LOG.error("", e);
		} finally {
			if (connection != null) {
				connection.close();
			}
		}

		//all new data has remove from old cache
		//remain record need to be delete
		for (Map.Entry<String, ArrayList<String>> oldCacheEntry : oldCache.entrySet()) {
			String cacheKey = oldCacheEntry.getKey();
			for (String fullKey : oldCacheEntry.getValue()) {
//					JSONObject changeLog = new JSONObject();
//					changeLog.put("version", this.version);
//					changeLog.put("action", "delete_fullKey");
//					changeLog.put("cacheKey", cacheKey);
//					changeLog.put("fullKey", fullKey);
//					changeLog.put("gateway", MathUtils.murmurHash(cacheKey.hashCode()) % xjoinInfo.getJoinInfo().getParallelism());
				Tuple5<Integer,String,String,String,Integer> changeLog=new Tuple5<Integer,String,String,String,Integer>(
					this.version,
					"delete_fullKey",
					cacheKey,
					fullKey,
					MathUtils.murmurHash(cacheKey.hashCode()) % xjoinInfo.getJoinInfo().getParallelism()
				);
				ctx.collect(changeLog);
			}

		}

		//send end message to all xjoin node
		for (int i = 0; i < xjoinInfo.getJoinInfo().getParallelism(); i++) {
//				JSONObject changeLogEnd = new JSONObject();
//				changeLogEnd.put("version", this.version);
//				changeLogEnd.put("gateway", i);
//				changeLogEnd.put("action", "end");
			Tuple5<Integer,String,String,String,Integer> changeLogEnd=new Tuple5<Integer,String,String,String,Integer>(
				this.version,
				"end",
				"",
				"",
				i
			);
			ctx.collect(changeLogEnd);
		}
		//replace oldCache
		cacheRef2.set(newCache);
		System.gc();
	}
}